package com.qfedu.common.mq.service.impl;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.qfedu.common.mq.entity.MqRetry;
import com.qfedu.common.mq.mapper.MqRetryMapper;
import com.qfedu.common.mq.service.MqRetryService;
import com.qfedu.common.mq.utils.RocketMqUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Classname MqRetryServiceImpl
 * @Description TODO
 * @Date 2025-09-30 11:45
 * @Created by 老任与码
 */
@Service
public class MqRetryServiceImpl implements MqRetryService {
    @Resource
    private MqRetryMapper mqRetryMapper;

    @Lazy
    @Resource
    private RocketMqUtils rocketMqUtils;

    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    @Override
    public void addMqRetry(MqRetry mqRetry) {
        mqRetryMapper.insert(mqRetry);
    }

    @Override
    public void resendMsgTask() {
        List<MqRetry> mqRetries = mqRetryMapper.selectList(Wrappers.lambdaQuery(MqRetry.class)
                .eq(MqRetry::getFlag, 0)
                .lt(MqRetry::getMaxNum, 3));
        if (mqRetries != null && mqRetries.size() > 0) {
            mqRetries.forEach(item -> {
                // 异步处理
                threadPoolExecutor.execute(() -> {
                    try {
                        rocketMqUtils.sendSyncMsg(item.getTopic(), item.getTag(), item.getMsg());
                        // 没有正常发送消息，修改flag=1,表示消息已处理
                        mqRetryMapper.update(null, Wrappers.lambdaUpdate(MqRetry.class)
                                .set(MqRetry::getFlag, 1)
                                .set(MqRetry::getMaxNum, item.getMaxNum() + 1)
                                .eq(MqRetry::getId, item.getId()));
                    } catch (Exception e) {
                        // 说明业务上重试失败，修改重试次数+1
                        mqRetryMapper.update(null, Wrappers.lambdaUpdate(MqRetry.class)
                                .set(MqRetry::getMaxNum, item.getMaxNum() + 1)
                                .eq(MqRetry::getId, item.getId()));
                    }
                });

            });
        }
    }
}
